import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  
public class Step4 {  

public static class MapClass extends Mapper<LongWritable, Text, DoubleWritable, Text> {  
    private Text word = new Text(); 
	
    @Override 
    public void map(LongWritable key, Text value, Context context) throws IOException,  InterruptedException {  
        try {
    	if (!value.toString().isEmpty()){
        	String lineFromText = value.toString();
            String[] lineSplited = lineFromText.split("\t");
            String[] pairSplited = lineSplited[0].split(" ");
            String firstPair = pairSplited[0];
            String secondPair = pairSplited[1];
            String words = firstPair+" "+secondPair;
            
            word.set(words);
        	double negativeSimilarity = -1*Double.parseDouble(lineSplited[1]);
        	context.write(new DoubleWritable(negativeSimilarity),word);
    	 }   	
		} catch (Exception e) {
				System.out.println("Error: In Step4, "+e);
		}
      }
  }
  
  public static class ReduceClass extends Reducer<DoubleWritable,Text,DoubleWritable,Text> { 
	private static int countKsimilarWords = 0;
	private int K;
		
    @Override 
    public void reduce(DoubleWritable key, Iterable<Text> values, Context context) throws IOException,  InterruptedException { 
    	List<String> valuesStr = new ArrayList<String>(); 
    	K = context.getConfiguration().getInt("K", 0);
    	
    	for (Text value: values){
    		if (countKsimilarWords < K){
    			valuesStr.add(value.toString());
    			countKsimilarWords++;
    		}
    		else
    			break;
    	}
    	
    	for (int i=0; i<valuesStr.size(); i++){
    		String[] splitWords = valuesStr.get(i).split(" ");
    		String similarityWords = ","+splitWords[0]+ " ," +splitWords[1];
    		double similarityValue = -1 * Double.parseDouble(key.toString());
    		context.write(new DoubleWritable(similarityValue), new Text(similarityWords));	
    	}
    }
  }
  
 public static void main(String[] args) throws Exception {
	Path input = new Path(args[0]);
	Path output = new Path(args[1]);
	int K = Integer.parseInt(args[2]);
    Configuration conf = new Configuration();
    conf.setInt("K", K);
    
    conf.set("mapred.max.split.size","5242880");
    
    Job job = new Job(conf, "Step4"); 
    job.setOutputKeyClass(DoubleWritable.class); 
    job.setOutputValueClass(Text.class); 
    job.setNumReduceTasks(1);
    job.setJarByClass(Step4.class); 
    job.setMapperClass(MapClass.class); 
    job.setReducerClass(ReduceClass.class); 
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    FileInputFormat.addInputPath(job, input); 
    FileOutputFormat.setOutputPath(job, output); 
    boolean flag = job.waitForCompletion(true); 
    if(!flag){
        System.err.println("Problem with Step4");
        return;
       }
  } 
}
  
